package com.bw.yk09;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.yk05.AsyncIOFunc4_4;
import com.bw.yk05.Mn5TM4Wide4;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Properties;
import java.util.TimeZone;
import java.util.concurrent.TimeUnit;

public class test7 {
    public static void main(String[] args) throws Exception {
        //流式环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为1
        env.setParallelism(1);

        //检查点
        env.enableCheckpointing(5000);
        //kafka读取数据
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop-single:9092");
        properties.setProperty("group.id", "test7");
        DataStream<String> order_wide_infoStream = env.addSource(new FlinkKafkaConsumer<>("dwm_order_info_bound_wide", new SimpleStringSchema(), properties));

        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        SingleOutputStreamOperator<OrderInfoWide> orderInfowideDS = order_wide_infoStream.map(new MapFunction<String, OrderInfoWide>() {
            @Override
            public OrderInfoWide map(String value) throws Exception {
                OrderInfoWide bean=JSON.parseObject(value, OrderInfoWide.class);
                return bean;
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy
                .<OrderInfoWide>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner((event, timestamp) -> {
                    try {
                        return sdf.parse(event.getOrder_info().getCreate_time()).getTime();
                    } catch (ParseException e) {
                        throw new RuntimeException(e);
                    }
                }));
        //orderInfoDS.print();

        SingleOutputStreamOperator<Order_org_stats> joinDS=orderInfowideDS.map(
                new MapFunction<OrderInfoWide, Order_org_stats>() {
                    @Override
                    public Order_org_stats map(OrderInfoWide orderInfoWide) throws Exception {
                        Order_org_stats bean=new Order_org_stats();
                        bean.setCreate_time_by_day(orderInfoWide.getOrder_info().getCreate_time().substring(0,10));
                        bean.setOrder_id(orderInfoWide.getOrder_info().getId());
                        bean.setOrg_id(orderInfoWide.getOrderOrgBound().getOrg_id());
                        bean.setOrder_order_num(1L);
                        bean.setOrder_amount(orderInfoWide.getOrder_info().getAmount());
                        return bean;
                    }
                }
        );

        KeyedStream<Order_org_stats, Tuple2<String, String>> keyedStream = joinDS.keyBy(new KeySelector<Order_org_stats, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> getKey(Order_org_stats value) throws Exception {
                return new Tuple2<>(
                        value.getCreate_time_by_day(),
                        value.getOrg_id());
            }
        });

        WindowedStream<Order_org_stats,Tuple2<String, String>, TimeWindow> windowedStream = keyedStream.window(TumblingEventTimeWindows.of(Time.days(1),Time.seconds(10)));


        SingleOutputStreamOperator<Order_org_stats> result = windowedStream.reduce(new ReduceFunction<Order_org_stats>() {
            @Override
            public Order_org_stats reduce(Order_org_stats value1, Order_org_stats value2) throws Exception {
                value1.setOrder_amount(value1.getOrder_amount().add(value2.getOrder_amount()));
                value1.setOrder_order_num(value1.getOrder_order_num() + value2.getOrder_order_num());
                return value1;
            }
        }, new WindowFunction<Order_org_stats, Order_org_stats, Tuple2<String, String>, TimeWindow>() {
            @Override
            public void apply(Tuple2<String, String> key, TimeWindow timeWindow, Iterable<Order_org_stats> iterable, Collector<Order_org_stats> collector) throws Exception {

                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

                //取出窗口的开始和结束时间
                long start = timeWindow.getStart();
                long end = timeWindow.getEnd();

                //取出数据
                Order_org_stats Order_org_stats = iterable.iterator().next();

                //设置窗口的开始和结束时间
                Order_org_stats.setStt(sdf.format(start));
                Order_org_stats.setEdt(sdf.format(end));

                //输出数据
                collector.collect(Order_org_stats);
            }
        });



        SingleOutputStreamOperator<Order_org_stats> asyncDS = AsyncDataStream.unorderedWait(result, new AsyncIOFuncyk09(), 1000, TimeUnit.MILLISECONDS, 100);


        asyncDS.print(">>>>>>>>>>>");
        asyncDS.addSink(new FlinkDWSSinkClickhouse2("insert into yk09sink02 values (?,?,?,?,?,?,?,?,?,?)"));


        env.execute();
    }
}
